#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include"muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include"../mqcommon/mq_helper.hpp"
#include"../mqcommon/mq_logger.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadPool.hpp"
#include"mq_consumer.hpp"
#include<iostream>
#include<mutex>
#include<condition_variable>
#include<unordered_map>

namespace xiang11mq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumerResponsePtr = std::shared_ptr<basicConsumerResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
class Channel
{
public:
    using ptr = std::shared_ptr<Channel>;
    Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
        :_cid(UUIDHelper::uuid())
        ,_conn(conn)
        ,_codec(codec)
    {}
    ~Channel()
    {
        //信道被关闭的时候 如果是消费者身份 需要取消订阅
        basicCancel();
    }
    //自己已经创建 向服务器发送创建信道请求
    bool openChannel()
    {
        openChannelRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        //等待服务器的响应
        basicCommonResponsePtr resp = waitResponse(rid);
        //返回
        return resp->ok();
    }
    //向服务器发送关闭信道请求 服务器关完 自己再关(上层关)
    void closeChannel()
    {
        closeChannelRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        //等待服务器的响应
        waitResponse(rid);
        //返回
        return;
    }
    //声明交换机
    bool declareExchange(const std::string &name, ExchangeType type, bool durable, bool auto_delete,
        google::protobuf::Map<std::string, std::string> &args)
    {
        //构造一个声明虚拟机的请求对象
        declareExchangeRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_exchange_name(name);
        req.set_exchange_type(type);
        req.set_durable(durable);
        req.set_auto_delete(auto_delete);
        req.mutable_args()->swap(args);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        basicCommonResponsePtr resp = waitResponse(rid);
        //返回
        return resp->ok();
    }
    //删除交换机
    void deleteExchange(const std::string &ename)
    {
        //构造一个声明虚拟机的请求对象
        deleteExchangeRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        waitResponse(rid);
        //返回
        return;
    }
    //声明消息队列
    bool declareQueue(const std::string &qname, bool qdurable, bool qexclusive, bool qauto_delete
        ,google::protobuf::Map<std::string, std::string> &qargs)
    {
        declareQueueRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(qname);
        req.set_durable(qdurable);
        req.set_auto_delete(qauto_delete);
        req.set_exclusive(qexclusive);
        req.mutable_args()->swap(qargs);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        basicCommonResponsePtr resp = waitResponse(rid);
        //返回
        return resp->ok();
    }
    //删除消息队列
    void deleteQueue(const std::string &qname)
    {
        deleteQueueRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(qname);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        waitResponse(rid);
        //返回
        return;
    }
    //添加绑定
    bool queueBind(const std::string &ename, const std::string &qname, const std::string &key)
    {
        queueBindRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(qname);
        req.set_exchange_name(ename);
        req.set_bind_key(key);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        basicCommonResponsePtr resp = waitResponse(rid);
        //返回
        return resp->ok();
    }
    //删除绑定
    void queueUnBind(const std::string &ename, const std::string &qname)
    {
        queueUnBindRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(qname);
        req.set_exchange_name(ename);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        waitResponse(rid);
        //返回
        return;
    }
    //消息发布
    void basicPublic(const std::string &ename, const BasicProperties *bp, const std::string &body)
    {
        basicPublishRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        DLOG("发布消息:%s", body.c_str());
        req.set_body(body);
        if(bp != nullptr)
        {
            req.mutable_properties()->set_id(bp->id());
            req.mutable_properties()->set_routing_key(bp->routing_key());
            req.mutable_properties()->set_deliver_mode(bp->deliver_mode());
        }
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        //DLOG("等待服务器收到消息发布请求");
        waitResponse(rid);
        //DLOG("收到服务器关于消息发布请求的回应");
        //返回
        return;
    }
    //消息确认
    void basicAck(const std::string &msg_id)
    {
        if(_consumer.get() == nullptr)
        {
            DLOG("消息确认时 找不到消费者信息");
        }
        basicAckRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(_consumer->subscribe_queue);
        req.set_message_id(msg_id);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        DLOG("等待服务器收到消息确认请求");        
        waitResponse(rid);
        DLOG("收到服务器关于消息确认请求的回应");
        //返回
        return;
    }
    //消息订阅
    bool basicConsume(const std::string &consumer_tag, const std::string &queue_name,
                        bool auto_ack, const ConsumerCallback &cb)
    {
        if(_consumer.get() != nullptr)
        {
            DLOG("一个信道最多有一个消费者描述 不能重复订阅");
        }
        DLOG("consumerclient: 消息订阅时候 consumer_tag: %s, queue_name: %s", consumer_tag.c_str(), queue_name.c_str());
        basicConsumerRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(queue_name);
        req.set_consumer_tag(consumer_tag);
        req.set_auto_ack(auto_ack);
        //向服务器发送请求 
        _codec->send(_conn, req);
        // 等待服务器的响应  
        basicCommonResponsePtr resp = waitResponse(rid);
        if(resp->ok() == false)
        {
            DLOG("添加订阅失败");
            return false;
        }
        _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
        // 返回
        return true;
    }
    //取消订阅
    void basicCancel()
    {
        basicCancelRequest req;
        std::string rid = UUIDHelper::uuid();
        req.set_rid(rid);        //requestID是唯一的
        req.set_cid(_cid);
        req.set_queue_name(_consumer->subscribe_queue);
        req.set_consumer_tag(_consumer->tag);
        //向服务器发送请求
        _codec->send(_conn, req);
        //等待服务器的响应
        waitResponse(rid);
        //返回
        return;
    }
    std::string cid() const
    {
        return _cid;
    }
public:
    //向连接开放的接口
    //连接收到基础响应后 向hash_map中添加响应
    void putBasicResponse(const basicCommonResponsePtr& resp)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _basic_resp.insert(std::make_pair(resp->rid(), resp));
        _cv.notify_all();

    }
    //连接收到消息推送后 需要通过信道找到对应的消费对象 通过回调函数进行消息处理
    void consume(const basicConsumerResponsePtr &resp)
    {
        if(_consumer.get() == nullptr)
        {
            DLOG("消息处理时 未找到订阅者消息");
            return;
        }
        if(_consumer->tag != resp->consumer_tag())
        {
            DLOG("收到推动消息的消费者标识%s:  与当前信道消费者标识%s 不一致", resp->consumer_tag().c_str(), _consumer->tag.c_str());
            return;
        }
        _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        
    }
private:
    //等待服务器响应
    basicCommonResponsePtr waitResponse(const std::string &rid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait(lock,[&rid, this]()
        {
            return _basic_resp.find(rid) != _basic_resp.end();
        } );
        //DLOG("客户端收到服务器响应rid: %s isok: %d", rid.c_str(), _basic_resp[rid]->ok());
        basicCommonResponsePtr basic_resp = _basic_resp[rid];
        _basic_resp.erase(rid);
        return basic_resp;
    }
private:
    std::string _cid;
    muduo::net::TcpConnectionPtr _conn;
    ProtobufCodecPtr _codec;                    //协议处理器一个就行 直接共享
    Consumer::ptr _consumer;                    //角色的描述
    std::mutex _mutex;
    std::condition_variable _cv;
    std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
};  


class ChannelManager
{
public:
    using ptr = std::shared_ptr<ChannelManager>;
    ChannelManager(){}
    Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
    {
        //uuid 唯一 不存在重复
        std::unique_lock<std::mutex> lock(_mutex);
        auto channel = std::make_shared<Channel>(conn, codec);
        _channels.insert(std::make_pair(channel->cid(), channel));
        return channel;
    }
    void remove(const std::string &cid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _channels.erase(cid);
    }
    Channel::ptr get(const std::string &cid)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        auto it = _channels.find(cid);
        if (it == _channels.end())
        {
            DLOG("client get channel failed! :%s", cid.c_str());
            return Channel::ptr();
        }
        return it->second;
    }
private:  
    std::mutex _mutex;
    std::unordered_map<std::string, Channel::ptr> _channels; //cid ptr
};

}


#endif